package net.yuyoo.uusession.core;

import java.io.Serializable;
import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

import net.yuyoo.uusession.support.JsonUtil;
import net.yuyoo.uusession.support.SeqUUIDUtil;

import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.coelacanth.core.membership.Endpoint;
import com.idroc.coelacanth.core.message.MessageDispatchHandler;
import com.idroc.coelacanth.core.message.support.StringMessage;
import com.idroc.coelacanth.core.message.support.SyncMessage;
import com.idroc.coelacanth.service.MessageService;

/**
 * 全局统一会话服务侍候器
 * @author yuyoo (yuyoo4j@163.com)
 * @teme 2010-6-4 下午03:12:54
 */
public class UUSessionServantInstance {
	
	private static Logger log = LoggerFactory.getLogger(UUSessionServantInstance.class);
	
	/**
	 * 调度服务
	 */
	private static ScheduledExecutorService scheduleService = Executors.newScheduledThreadPool(3);
	
	/**
	 * 配置对象
	 */
	private UUSessionConfig config = null;
	 	
	/**
	 * 会话服务侍候器 元数据
	 */
	private EndpointMeta meta = null;
		
	
	/**
	 * ConsistentHash管理 后端 Memcached 代理集合
	 */
	private ConsistentHashRing<MemcachedClientProxy> locator = new ConsistentHashRing<MemcachedClientProxy>();
	
	/**
	 * 会话服务侍候器 集合
	 */
	private SortedMap<Long, EndpointMeta> metas = new TreeMap<Long, EndpointMeta>(); 
	
	
	/**
	 * 消息服务
	 */
	private MessageService msgService = null;
	
	/**
	 * 同步消息处理器
	 */
	private SyncMessageHandler handler = null;
	
	/**
	 * 构造方法
	 * @param config -- 配置对象
	 */
	public UUSessionServantInstance(UUSessionConfig config) { 
		this.config = config;
		this.initialize();
	}
	
	/**
	 * 构造方法
	 */
	public UUSessionServantInstance() {
		this(new UUSessionConfig("/uusession.config.json"));
	}
	
	/**
	 * 初始化
	 */
	private void initialize() {

		config.setTimingReload(false);
		config.initialize();
		
		meta = new EndpointMeta();
		meta.no = JsonUtil.toLong(config.getValue("instanceNo"));		
		String pn = config.getValue("poolName");
		meta.poolName = (null != pn) ? pn : meta.poolName;
		int timeout = JsonUtil.toInt(config.getValue("sessionTimeout"));
		meta.sessionTimeout = (0 != timeout) ? timeout : meta.sessionTimeout;		
		meta.appData = config.getValue("appData"); 
		metas.put(meta.no, meta);		
		
		initializeMemcached();
		initializeMessageSystem();
		
		String ref = config.getValue("startupReference");
		if (null != ref && !"".equals(ref)) {
			meta.startupReference = ref;
			initializeStartupReference(ref);
		}		
		initializeDaemonTask();
	}
	
	/**
	 * 初始化消息系统
	 */
	private void initializeMessageSystem() {
		
		handler = new SyncMessageHandler();		
		int port = JsonUtil.toInt(config.getValue("instancePort"));
		String ip = config.getValue("instanceIp");
		Endpoint ep = new Endpoint();
		ep.setPort(port);
		if (null == ip || "".equals(ip)) {
			try {
				InetAddress ia = InetAddress.getLocalHost(); 	
				meta.endpoint = new Endpoint(ia.getHostAddress(), port); 
			} catch (Exception ex) {
				throw new IllegalStateException("获取instanceIp参数异常", ex);
			}
		} else {
			meta.endpoint = new Endpoint(ip, port);
			ep.setHost(ip);
		}
		log.debug("instanceIp:{}, instancePort:{}", ip, port);	
		
		msgService = new MessageService(ep);
		msgService.start();
		MessageDispatchHandler sd = msgService.getServerMessageFactory().getMessageDispatcher();
		sd.addUpMessageHandler(SyncMessage.class, handler); 
		
		StringMessageHandler smh = new StringMessageHandler();
		sd.addUpMessageHandler(StringMessage.class, smh); 
		
		MessageDispatchHandler cd = msgService.getClientMessageFactory().getMessageDispatcher();
		cd.addUpMessageHandler(SyncMessage.class, handler); 
		cd.addUpMessageHandler(StringMessage.class, smh); 
		
		log.debug("init msg system");
	}
	
	private void initializeMemcached() {
		
		Map<String, String> ms = config.getValue("backendMemcacheds"); 
		for (String key : ms.keySet()) {
			String val = ms.get(key);
			MemcachedClientProxy proxy = new MemcachedClientProxy(val);
			locator.addNode(Long.valueOf(key), proxy);
		}
		log.debug("memcacheds:{}", ms);
	}
	
	/**
	 * 初始化启动引用端点
	 * @param ref -- 启动引用端点 
	 */
	@SuppressWarnings("unchecked")
	private void initializeStartupReference(String ref) {
		
		String[] conn = ref.trim().split(":");
		log.debug("startupReference:{}", ref);
		Endpoint to = new Endpoint(conn[0], Integer.parseInt(conn[1]));
		Map<String, Object> params = new HashMap<String, Object>();
		params.put("method", "instance.members.list");
		Map<String, Object> rtv = exchangeData(to, params);
		log.debug("response:{}", rtv);
		Map<String, String> bms = (Map<String, String>) rtv.get("backendMemcacheds");
		if (null != bms) {
			for (String key : bms.keySet()) {
				String val = bms.get(key);
				MemcachedClientProxy proxy = new MemcachedClientProxy(val);
				locator.addNode(Long.valueOf(key), proxy);
			}
		}
		Map<String, Object> ms = (Map<String, Object>)rtv.get("metas");
		if (null != ms) {
			for (String key : ms.keySet()) {
				Map<String, Object> m = (Map<String, Object>) ms.get(key);
				EndpointMeta item = toMetaBean(m);
				metas.put(Long.valueOf(key), item);
				registerTo(meta, item);
			}
		}
		log.debug("updated backendMemcacheds:{}", bms);
		log.debug("updated metas:{}", metas);
	}
	
	private void registerTo(EndpointMeta meta, EndpointMeta remote) {
		
		log.debug("向另外一个会话服务侍候器注册当前端点");
		log.debug("from:{}, to:{}", meta, remote);
		try {
			Map<String, Object> params = new HashMap<String, Object>();
			params.put("method", "instance.meta.register");
			params.put("from", meta);
			Map<String, Object> rec = exchangeData(remote.endpoint, params);
			log.debug("rec:{}", rec);
			EndpointMeta target = toMetaBean((Map<String, Object>) rec.get("target"));
			log.debug("target meta:{}", target);
		} catch (Exception ex) {
			log.error("向另外一个会话服务侍候器注册失败.", ex);
		}
	}
	
	/**
	 * meta信息从map转成bean
	 * @param meta -- meta map
	 * @return -- meta bean
	 */
	@SuppressWarnings("unchecked")
	private EndpointMeta toMetaBean(Map<String, Object> meta) {
		
		EndpointMeta m = new EndpointMeta();
		m.no = JsonUtil.toLong(meta.get("no"));
		Map<String, Object> ep = (Map<String, Object>) meta.get("endpoint");
		Endpoint endpoint = new Endpoint();
		endpoint.setHost((String) ep.get("host"));
		endpoint.setPort(JsonUtil.toInt(ep.get("port")));
		m.endpoint = endpoint;
		m.appData = (Map<String, Object>) meta.get("appData");
		m.heartbeatCount = JsonUtil.toLong(meta.get("heartbeatCount"));
		m.lastTimestamp = JsonUtil.toLong(meta.get("lastTimestamp"));
		m.poolName = (String) meta.get("poolName");
		m.sessionTimeout = JsonUtil.toLong(meta.get("sessionTimeout"));
		m.startupReference = (String) meta.get("startupReference");
		m.timestamp = JsonUtil.toLong(meta.get("timestamp"));
		return m;
	}
	
	/**
	 * 初始化任务
	 */
	private void initializeDaemonTask() {
		
		final long period = getUpdatePeriod();
		log.debug("daemon task period:{}", period);
		Runnable task = new Runnable() {
			public void run() {
				log.debug("心跳同步任务开始......");
				try {
					for (Long key : metas.keySet()) {
						EndpointMeta m = metas.get(key);
						log.debug("before update meta:{}", m);
						try {
							if (m == meta) { // 当前 服务侍候器
								m.heartbeatCount++;
								m.lastTimestamp = m.timestamp;
								m.timestamp = System.currentTimeMillis();
							} else {
								updateEndpointMeta(m);
							}
						} catch (Exception iex) {
							log.info("与端点[{}]心跳同步异常", m);
							log.debug("心跳同步异常信息", iex);
						}
						log.debug("after update meta:{}", m);
					}
				} catch (Exception ex) {
					log.info("心跳同步任务异常", ex);
				} 
				log.debug("心跳同步任务结束");
			}			
		}; 
		scheduleService.scheduleAtFixedRate(task, 0, period, TimeUnit.MILLISECONDS);
	}
	
	/**
	 * 更新端点元数据
	 * @param now -- 需要更新的端点
	 */
	@SuppressWarnings("unchecked")
	private void updateEndpointMeta(EndpointMeta now) {
		
		Map<String, Object> params = new HashMap<String, Object>();
		params.put("method", "instance.meta.latest");
		params.put("target", now);
		Map<String, Object> rec = exchangeData(now.endpoint, params);
		log.debug("rec:{}", rec);
		EndpointMeta target = toMetaBean((Map<String, Object>) rec.get("target"));  
		now.heartbeatCount = target.heartbeatCount;
		now.timestamp = target.timestamp;
		now.lastTimestamp = target.lastTimestamp;
	}
	
	/**
	 * 获取服务侍候器的名称
	 * @return -- 名称
	 */
	public String getName() {
		return this.config.getValue("instanceName");
	}
	
	/**
	 * 根据sessionId获取Memcached客户端代理
	 * @param sessionId -- 会话Id
	 * @return -- Memcached客户端代理
	 */
	public 	MemcachedClientProxy getBackendProxy(String sessionId) {
		return locator.locate(sessionId);
	}
	
	public long getUpdatePeriod() {
		return (long) (meta.sessionTimeout * 0.85f);
	}
	 
	public EndpointMeta getMeta(Long no) {
		return metas.get(no);
	}
	
	public EndpointMeta getMeta() {
		return meta;
	}
	
	/**
	 * 根据会话服务侍候器的编号更新其元数据
	 * @param no -- 会话服务侍候器的编号
	 */
	public void updateEndpointMeta(final long no) {

		EndpointMeta target = metas.get(Long.valueOf(no));
		if (null != target && target != meta) {
			updateEndpointMeta(target);
		}				
	}
	
	/**
	 * 根据会话服务侍候器的编号更新其元数据, 异常执行(提交后台线程执行)
	 * @param no -- 会话服务侍候器的编号
	 */
	public void asyncUpdateEndpointMeta(final long no) {
		Runnable task = new Runnable() {
			public void run() {
				updateEndpointMeta(no);
			}
		};
		scheduleService.execute(task);
	}
	
	/**
	 * 交换数据
	 * @param to -- 目标端点
	 * @param params -- 交换的数据
	 * @return -- 交换后响应的数据
	 */
	@SuppressWarnings("unchecked")
	public Map<String, Object> exchangeData(Endpoint to, Map<String, Object> params) {
		
		String content = JsonUtil.toJson(params);
		StringMessage sm = new StringMessage();
		sm.setContent(content);
		StringMessage rec = syncSendMessage(to, sm, 10000);
		log.debug("response json:{}", rec.getContent());
		Map<String, Object> rtv = JsonUtil.toBean(rec.getContent(), Map.class);
		return rtv;
	}
	
	@Override
	public String toString() {

		StringBuilder sb = new StringBuilder();
		sb.append("UUSessionServantInstance{");
		sb.append("meta:").append(meta);
		sb.append("}");
		return sb.toString();
	}
	
	/**
	 * 同步发送消息
	 * @param to -- 目标端点
	 * @param msg -- 发送字符串消息
	 * @param timeout -- 等待超时,单位:毫秒
	 * @return -- 响应字符串消息
	 */
	public StringMessage syncSendMessage(final Endpoint to, StringMessage msg, long timeout) {
		
		String id = SeqUUIDUtil.toSequenceUUID();
		final SyncMessage sm = new SyncMessage();
		sm.setId(id);
		sm.setContent(msg); 
		msgService.sendMessage(to, sm); 
		BlockingQueue<StringMessage>  bq = new SynchronousQueue<StringMessage>();
		handler.waits.put(id, bq);
		StringMessage receive = null;
		try {
			receive = bq.poll(timeout, TimeUnit.MILLISECONDS);
		} catch (Exception ex) {
			handler.waits.remove(id);
			String errorMsg = "等待reqId" + id + "的响应异常";
			throw new RuntimeException("同步发送字符串消息超时异常, 异常:" + errorMsg, ex);
		}
		if (null == receive) {
			handler.waits.remove(id); 
			throw new RuntimeException("reqId[" + id + "]没有响应消息");
		}
		return receive;
	}
	
	/**
	 * 同步消息处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @teme 2010-6-5 上午05:49:24
	 */
	@SuppressWarnings("unchecked")
	private class SyncMessageHandler extends SimpleChannelHandler {
				
		Map<String, BlockingQueue> waits = new ConcurrentHashMap<String, BlockingQueue>(300);

		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception { 
						
			SyncMessage msg = (SyncMessage) e.getMessage();
			StringMessage sm = (StringMessage) msg.getContent(); 
			if (msg.isReq()) {
				Map<String, Object> params = JsonUtil.toBean(sm.getContent(), Map.class);
				String method = (String) params.get("method");
				if ("instance.members.list".equals(method)) {
					log.debug("接受另外的端点的查询端点列表集合的请求");
					Map<String, String> backends = new HashMap<String, String>();
					for (Long key : locator.nodes.keySet()) {
						MemcachedClientProxy val = locator.nodes.get(key);
						backends.put(key.toString(), val.getEndpoints());
					}
					Map<String, Object> res = new HashMap<String, Object>();
					res.put("backendMemcacheds", backends);
					res.put("metas", metas);
					String json = JsonUtil.toJson(res);
					sm.setContent(json);
					msg.setReq(false);
					ctx.getChannel().write(msg);
					log.debug("写同步响应消息:{}", msg);
				} else if ("instance.meta.latest".equals(method)) {
					log.debug("接受另外的端点的更新当前端点的元数据的请求");
					Map<String, Object> res = new HashMap<String, Object>();
					res.put("target", meta);
					String json = JsonUtil.toJson(res);
					sm.setContent(json);
					msg.setReq(false);
					ctx.getChannel().write(msg);
				} else if ("instance.meta.register".equals(method)) {
					EndpointMeta from = toMetaBean((Map<String, Object>) params.get("from"));
					log.debug("接受另外端点[{}]的注册请求", from);
					metas.put(Long.valueOf(from.no), from);
					Map<String, Object> res = new HashMap<String, Object>();
					res.put("target", meta);
					String json = JsonUtil.toJson(res);
					sm.setContent(json);
					msg.setReq(false);
					ctx.getChannel().write(msg);
				}
			} else { 
				log.debug("捕获同步响应消息:{}", msg);
				String reqId = msg.getId();
				BlockingQueue bq = waits.remove(reqId);
				if (null != bq) {
					bq.offer(sm);
				} else {
					log.info("no waiting request:{}", reqId);
				}
			}
		}
	}
	
	/**
	 * 字符串消息处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @teme 2010-6-5 上午03:50:49
	 */
	private class StringMessageHandler  extends SimpleChannelHandler {
				
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception { 
			StringMessage msg = (StringMessage) e.getMessage();
			log.debug("StringMessage:{}", msg); 
		}
	}
	
	/**
	 * 会话服务侍候器 元数据
	 * @author yuyoo (yuyoo4j@163.com)
	 * @teme 2010-6-6 上午12:58:12
	 */
	@SuppressWarnings("serial")
	public class EndpointMeta implements Serializable {
		
		/**
		 * 编号
		 */
		public long no = -1;
		
		/**
		 * 端点
		 */
		public Endpoint endpoint = null;
		
		/**
		 * 心跳次数(从0开始计数)
		 */
		public long heartbeatCount = 0;

		/**
		 * 心跳发生时间点(java时间戳)
		 */
		public long timestamp = 0; 
		
		/**
		 * 倒数第二次心跳的发生时间点(java时间戳)
		 */
		public long lastTimestamp = 0;

		/**
		 * 会话池 名称
		 */
		public String poolName = "yuyoo"; // default pool name: 'yuyoo'
		
		/**
		 * 会话超时设置
		 */
		public long sessionTimeout = 1800000; // default 1800s
		
		/**
		 * 启动时引用 端点信息
		 */
		public String startupReference = null;
		
		/**
		 * 应用附加数据(辅助信息)
		 */
		public Map<String, Object> appData = null;
		
		@Override
		public String toString() { 
			
			StringBuilder sb = new StringBuilder();
			sb.append("\nEndpointMeta{\n");
			sb.append(" no:").append(no).append(",\n");
			sb.append(" endpoint:").append(endpoint).append(",\n");
			sb.append(" heartbeatCount:").append(heartbeatCount).append(",\n");
			sb.append(" timestamp:").append(timestamp).append(",\n");
			sb.append(" lastTimestamp:").append(lastTimestamp).append(",\n");
			sb.append(" poolName:").append(poolName).append(",\n");
			sb.append(" sessionTimeout:").append(sessionTimeout).append(",\n");
			sb.append(" startupReference:").append(startupReference).append(",\n");
			sb.append(" appData:").append(appData); 
			sb.append("}");
			return sb.toString();
		}
	} 
}
